package com.galeno.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.File;
import java.io.RandomAccessFile;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2321:06
 */
public class MySource extends RichSourceFunction implements CheckpointedFunction {
    private boolean flag=true;
    private long offset=0;
    private String path;
    private transient ListState<Long> listState;

    public MySource(String path){
        this.path=path;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        listState.clear();
        listState.add(offset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> longListStateDescriptor = new ListStateDescriptor<>("offset-state", Long.class);
        listState=  context.getOperatorStateStore().getListState(longListStateDescriptor);
        /**
         * 判断是否恢复完成
         */
        if (context.isRestored()){
            for (Long offset : listState.get()) {
                this.offset=offset;
            }


        }



    }

    @Override
    public void run(SourceContext ctx) throws Exception {
        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
        RandomAccessFile raf = new RandomAccessFile(path + File.separator + indexOfThisSubtask + ".txt", "r");
        raf.seek(offset);
        while (flag){
            String line = raf.readLine();
            if (line!=null){
                synchronized (ctx.getCheckpointLock()){
                    ctx.collect(indexOfThisSubtask+"-----"+line);
                    offset=raf.getFilePointer();//更新偏移量
                }

            }else {
                Thread.sleep(1000);
            }
        }


    }

    @Override
    public void cancel() {
            flag=false;
    }
}
